﻿using Confluent.Kafka;
using Confluent.Kafka.Admin;
using System;
using System.Collections.Generic;
using System.Threading.Tasks;

namespace AdminClient
{
    class Program
    {
        static string ToString(int[] array) => $"[{string.Join(",", array)}]";

        static void ListGroups(string bootstrapServers)
        {
            using(var adminClient= new AdminClientBuilder(new AdminClientConfig { BootstrapServers = bootstrapServers }).Build())
            {
                var groups = adminClient.ListGroups(TimeSpan.FromSeconds(10));
                Console.WriteLine($"Consumer Groups:");
                foreach (var g in groups)
                {
                    Console.WriteLine($"    Group: {g.Group} {g.Error} {g.State}");
                    Console.WriteLine($"    Broker: {g.Broker.BrokerId} {g.Broker.Host} {g.Broker.Port}");
                    Console.WriteLine($"    Protocol: {g.ProtocolType} {g.Protocol}");
                    Console.WriteLine($"    Members:");
                    foreach (var m in g.Members)
                    {
                        Console.WriteLine($"    {m.MemberId} {m.ClientId} {m.ClientHost}");
                        Console.WriteLine($"    Metadata: {m.MemberMetadata.Length} bytes");
                        Console.WriteLine($"    Assignment: {m.MemberAssignment.Length} bytes");
                    }
                }
            }
        }

        static void PrintMetadata(string bootstrapServers)
        {
            using (var adminClient = new AdminClientBuilder(new AdminClientConfig { BootstrapServers = bootstrapServers }).Build())
            {
                var meta = adminClient.GetMetadata(TimeSpan.FromSeconds(20));
                Console.WriteLine($"{meta.OriginatingBrokerId} {meta.OriginatingBrokerName}");
                meta.Brokers.ForEach(broker =>
                    Console.WriteLine($"Broker: {broker.BrokerId} {broker.Host}:{broker.Port}"));
                meta.Topics.ForEach(topic => {
                    Console.WriteLine($"Topic: {topic.Topic} {topic.Error}");
                    topic.Partitions.ForEach(partition => {
                        Console.WriteLine($"  Partition: {partition.PartitionId}");
                        Console.WriteLine($"    Replicas: {ToString(partition.Replicas)}");
                        Console.WriteLine($"    InSyncReplicas: {ToString(partition.InSyncReplicas)}");
                    });
                });
            }
        }

        static async Task CreateTopicAsync(string bootstrapServers, string topicName)
        {
            using(var adminClient = new AdminClientBuilder(new AdminClientConfig { BootstrapServers = bootstrapServers }).Build())
            {
                try
                {
                    await adminClient.CreateTopicsAsync(new TopicSpecification[] {
                        new TopicSpecification { Name = topicName, ReplicationFactor = 1, NumPartitions = 1 }
                    });
                }
                catch(CreateTopicsException e)
                {
                    Console.WriteLine($"An error occured creating topic {e.Results[0].Topic}: {e.Results[0].Error.Reason}");
                }
            }
        }

        static async Task DeleteTopicAsync(string bootstrapServers, string topicName)
        {
            using (var adminClient = new AdminClientBuilder(new AdminClientConfig { BootstrapServers = bootstrapServers }).Build())
            {
                try
                {
                    await adminClient.DeleteTopicsAsync(new string[] { topicName });
                }
                catch (CreateTopicsException e)
                {
                    Console.WriteLine($"An error occured creating topic {e.Results[0].Topic}: {e.Results[0].Error.Reason}");
                }
            }
        }

        static async Task CreatePartitionAsync(string bootstrapServers, string topicName, int partition)
        {
            using (var adminClient = new AdminClientBuilder(new AdminClientConfig { BootstrapServers = bootstrapServers }).Build())
            {
                try
                {
                    PartitionsSpecification specification = new PartitionsSpecification();
                    specification.Topic = topicName;
                    specification.IncreaseTo = partition;
                    List<PartitionsSpecification> partitions = new List<PartitionsSpecification>();
                    partitions.Add(specification);
                    await adminClient.CreatePartitionsAsync(partitions);
                }
                catch (CreatePartitionsException e)
                {
                    Console.WriteLine($"An error occured creating partition {e.InnerException?.ToString()}: {e.Results[0].Error.Reason}");
                }
            }
        }

        static async Task Main(string[] args)
        {
            if(args.Length < 2)
            {
                Console.WriteLine("usage: .. <bootstrapServers> <list-groups|metadata|library-version|create-topic|delete-topic|create-partition> [topic-name] [partionId]");
                System.Environment.Exit(1);
            }

            switch (args[1])
            {
                case "library-version":
                    Console.WriteLine($"librdkafka Version: {Library.VersionString} ({Library.Version:X})");
                    Console.WriteLine($"Debug Contexts: {string.Join(", ", Library.DebugContexts)}");
                    break;
                case "list-groups":
                    ListGroups(args[0]);
                    break;
                case "metadata":
                    PrintMetadata(args[0]);
                    break;
                case "create-topic":
                    await CreateTopicAsync(args[0], args[2]);
                    break;
                case "delete-topic":
                    await DeleteTopicAsync(args[0], args[2]);
                    break;
                case "create-partition":
                    await CreatePartitionAsync(args[0], args[2], Convert.ToInt32(args[3]));
                    break;
                default:
                    Console.WriteLine($"unkown command: {args[1]}");
                    break;
            }
        }
    }
}
